package com.hkbigdata.flink.task;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hkbigdata.flink.domain.OrderProcess;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.time.Duration;
import java.util.Date;

import static com.hkbigdata.flink.utils.Constants.*;

@Slf4j
public class C02_WindowOrder {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 从Kafka读：新Source架构
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
//                .setBootstrapServers("hadoop102:9093") // 指定kafka节点的地址和端口
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS) // 指定kafka节点的地址和端口
//                .setGroupId("task")  // 指定消费者组的id
                .setTopics(KAFKA_TOPIC_CDC)   // 指定消费的 Topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value
                .setStartingOffsets(OffsetsInitializer.earliest())  // flink消费kafka的策略
                .build();



        WatermarkStrategy<OrderProcess> orderProcessWatermarkStrategy = WatermarkStrategy
                .<OrderProcess>forBoundedOutOfOrderness(Duration.ofSeconds(FLINK_WINDOWS_DELAY_TIME))
                .withTimestampAssigner((event, timestamp) -> event.getCreateTime());

        SingleOutputStreamOperator<String> orderDS = env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkasource")
                .flatMap(new FlatMapFunction<String, OrderProcess>() {
                             @Override
                             public void flatMap(String value, Collector<OrderProcess> out) throws Exception {
                                 //System.out.println(value);
                                 try {
                                     ObjectMapper objectMapper = new ObjectMapper();
                                     JsonNode jsonNode = objectMapper.readTree(value);
                                     OrderProcess orderProcess = new OrderProcess();
                                     String table = jsonNode.get("source").get("table").asText();
                                     String operation = jsonNode.get("op").asText();
                                     long time = jsonNode.get("after").get("create_time").asLong();
                                     if ("ord_order".equals(table) && "c".equals(operation)) {
                                         orderProcess.setTable(table);
                                         orderProcess.setOperation(operation);
                                         orderProcess.setCreateTime(time);
//                                         System.out.println("数据2=" + orderProcess + ",recordTs=" + new Date(time));
                                         out.collect(orderProcess);
                                     }
                                 } catch (Exception e) {
//                                     e.printStackTrace();
                                     System.out.println("异常为：" + e.getMessage() + " 异常数据为：" + value);
                                 }
                             }
                         }
                )
                .assignTimestampsAndWatermarks(orderProcessWatermarkStrategy) // 自定义时间戳分配器和水印生成器
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(FLINK_WINDOWS))) // 使用滑动窗口，窗口大小10s
                .apply(new AllWindowFunction<OrderProcess, String, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<OrderProcess> orders, Collector<String> out) throws Exception {
                        Long count = 0L;
                        for (OrderProcess order : orders) {

                            count++;
                            System.out.println("时间为" + new Date(window.getStart()) + "<->" + new Date(window.getEnd()) + " 收集的数据- " + order + new Date(order.getCreateTime()) + " " + count);
                        }
                        out.collect(window.getStart() + ":" + count);
                    }
                });
//                .print();
        orderDS.print();

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                // 指定 kafka 的地址和端口
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS)
                // 指定序列化器：指定Topic名称、具体的序列化
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(KAFKA_TOPIC_ORDER_WINDOW)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                // 写到kafka的一致性级别： 精准一次、至少一次
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 如果是精准一次，必须设置 事务的前缀
                .setTransactionalIdPrefix("task-process2-")
                // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                .build();

        orderDS.sinkTo(kafkaSink);
        env.execute();
    }
}
